#include "rtmp_reader.h"
#include "nalu.h"

CRtmpReader::CRtmpReader()
{
    mState = STAT_IDLE;
    mHandle = NULL;
    mIsRunning = false;
    mPacketHandler = NULL;
    mServer[0] = 0;
    
    // 创建线程
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    pthread_create(&mThread, &attr, thread_func, this);
}

CRtmpReader::~CRtmpReader()
{
    LOGE("~~CRtmpReader free ..");
    if (mThread != 0) {
        mIsRunning = false;
        for (int i = 0; i < 10; i++) {
            if (mThread == 0)
                break;
            usleep(1000 * 50);
        }
    }
}


int CRtmpReader::open(const char* serverip, const char* id, int channel, int sndtimeout, int rcvtimeout)
{
    strcpy(mID, id);
    mChannel = channel;
    mSndTimeout = sndtimeout;
    mRcvTimeout = rcvtimeout;
    
    if (mState == STAT_RTMP_PAUSE_WAIT) {
        if (strcmp(mServer, serverip) == 0) {
            char stream[64];
            srs_rtmp_pause_stream(mHandle, 0);
            sprintf(stream, "%s%d", id, channel);
            srs_rtmp_swap_stream(mHandle, stream);
            mState = STAT_RTMP_RECEIVE;
            LOGI("Resume rtmp connection.");
            return 0;
        }
        rtmpClose();
    }
    strcpy(mServer, serverip);
    
    mWaitTime = 0;
    mState = STAT_RTMP_OPEN;
    return 0;
}

void CRtmpReader::close(int seconds)
{
    if (seconds == 0 || mState != STAT_RTMP_RECEIVE) {
        mState = STAT_RTMP_CLOSE;
    }
    else {
        mWaitTime = sys_gettime() + seconds * 1000;
        mState = STAT_RTMP_PAUSE;
    }
}

int CRtmpReader::swapStream(const char* stream_name)
{
    if (mState == STAT_RTMP_RECEIVE) {
        if (srs_rtmp_swap_stream(mHandle, stream_name) == 0)
            return 0;
    }
    return ERR_RTMP_ERROR;
}

int CRtmpReader::saveStream(const char* stream_name, const char* serial_no)
{
    if (mState == STAT_RTMP_RECEIVE) {
        if (srs_rtmp_save_stream(mHandle, stream_name, serial_no) == 0)
            return 0;
    }
    return ERR_RTMP_ERROR;
}

int CRtmpReader::pauseStream(int pause)
{
    if (mState == STAT_RTMP_RECEIVE) {
        if (srs_rtmp_pause_stream(mHandle, pause) == 0)
            return 0;
    }
    return ERR_RTMP_ERROR;
}

bool CRtmpReader::isConnected()
{
    return (mHandle != NULL && mState == STAT_RTMP_RECEIVE);
}

void CRtmpReader::onThread()
{
    int64_t _now;
    
    mIsRunning = true;
    while (mIsRunning) {
        switch (mState) {
            case STAT_IDLE:
            case STAT_ERROR:
                usleep(1000 * 50);
                break;
            case STAT_RTMP_OPEN:
                _now = sys_gettime();
                if (_now >= mWaitTime) {
                    rtmpClose();
                    mHandle = rtmpOpen();
                    if (mHandle != NULL)
                        mState = STAT_RTMP_RECEIVE;
                    else
                        mWaitTime = _now + 1000;
                }
                break;
            case STAT_RTMP_CLOSE:
                rtmpClose();
                mState = STAT_IDLE;
                break;
            case STAT_RTMP_RECEIVE:
                rtmpReceive();
                break;
            case STAT_RTMP_PAUSE:
                srs_rtmp_pause_stream(mHandle, 1);
                mState = STAT_RTMP_PAUSE_WAIT;
                break;
            case STAT_RTMP_PAUSE_WAIT:
                _now = sys_gettime();
                if (_now > mWaitTime) {
                    rtmpClose();
                    mState = STAT_IDLE;
                }
                else {
                    usleep(1000 * 10);
                }
                break;
        }
    }
    
    rtmpClose();
    mThread = 0;
}

srs_rtmp_t CRtmpReader::rtmpOpen()
{
    char url[255];
    int ret;
    int64_t start_time, conn_time, ok_time;
    
    sprintf(url, "rtmp://%s/game/%s%d", mServer, mID, mChannel);
    LOGI("opening %s", url);
    
    start_time = sys_gettime();
    srs_rtmp_t rtmp = srs_rtmp_create(url);
    if (rtmp) {
        if ((ret = srs_rtmp_handshake(rtmp)) != 0) {
            LOGE("srs_rtmp_handshake ERROR: %d", ret);
            srs_rtmp_destroy(rtmp);
            return NULL;
        }
        conn_time = sys_gettime();
        
        srs_rtmp_set_timeout(rtmp, 3000, 2000);
        if ((ret = srs_rtmp_connect_app(rtmp)) != 0) {
            LOGE("srs_rtmp_connect_app ERROR: %d", ret);
            srs_rtmp_destroy(rtmp);
            return NULL;
        }
        else if ((ret = srs_rtmp_play_stream(rtmp)) != 0) {
            LOGE("srs_rtmp_play_stream ERROR: %d", ret);
            srs_rtmp_destroy(rtmp);
            return NULL;
        }
        ok_time = sys_gettime();
        LOGI("[STATIS] CONNECT SUCCESS, connect server used %d ms, connect stream used %d ms, total used %d ms.",
             (int)(conn_time - start_time), (int)(ok_time - conn_time), (int)(ok_time - start_time));
        
        srs_rtmp_set_timeout(rtmp, mRcvTimeout, mSndTimeout);
    }
    
    //mLastRecvTime = sys_gettime();
    LOGI("rtmp connected.");
    
    return rtmp;
}

void CRtmpReader::rtmpClose()
{
    if (mHandle) {
        srs_rtmp_destroy(mHandle);
        mHandle = NULL;
    }
}

void CRtmpReader::rtmpReceive()
{
    int64_t _now;
    char* data;
    int size, ret;
    char type;
    u_int32_t timestamp;
    
    if ((ret = srs_rtmp_read_packet(mHandle, &type, &timestamp, &data, &size)) == 0) {
        if (type != SRS_RTMP_TYPE_VIDEO) {
            payload_freep(data);
            return;
        }
        
        mRecvBytes += size;
        mRecvPackets++;
        // mLastRecvTime = sys_gettime();
        
        if (mPacketHandler && (data[9] & 0x1f) != NALU_TYPE_AUD)
            mPacketHandler(data, size, mPacketHandlerParam);
        else
            payload_freep(data);
    }
    else {
        LOGE("read video packet error! (%d)", ret);
        
        // 处理视频出错，需要关闭重连视频
        rtmpClose();
        mWaitTime = 0;
        if (mState == STAT_RTMP_RECEIVE)
            mState = STAT_RTMP_OPEN;
    }
}

void CRtmpReader::getReceivedInfo(int* rcvbytes, int* rcvpkts)
{
    *rcvbytes = mRecvBytes;
    *rcvpkts = mRecvPackets;
    mRecvBytes = 0;
    mRecvPackets = 0;
}

